package com.bluedot.queue;

import com.bluedot.message.Message;

import java.util.Deque;
import java.util.LinkedList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 顶级抽象队列类<br/>
 * 定义并且实现了4个实现类都会使用到的属性及方法
 */
public abstract class AbstractQueue {

    protected final ReentrantLock lock;

    protected final Condition fullWaitSet;

    protected final Condition emptyWaitSet;

    protected final Deque<Message> queue;

    private static final Integer QUEUE_SIZE = 100;

    protected AbstractQueue() {
        this.lock = new ReentrantLock();
        this.fullWaitSet = lock.newCondition();
        this.emptyWaitSet = lock.newCondition();
        this.queue = new LinkedList<>();
    }

    /**
     * 将消息入队,不设置超时时间,当队列为满时当前线程进入emptyWaitSet等待队列进行等待,队列一直为满则一直等待,直到有别的线程唤醒<br/>
     * 将消息入队前先获取锁,获取失败则当前线程进入阻塞队列且被interrupt,直到其他线程释放锁当前线程再进行锁竞争<br/>
     * 消息入队成功时要随机唤醒一个在emptyWaitSet中等待的线程
     *
     * @param message 要入队的消息
     * @return 是否入队成功, 本方法队列为满时会一直等待, 所以一定成功
     */
    public boolean push(Message message) {
        lock.lock();
        try {
            while (queue.size() == QUEUE_SIZE) {
                try {
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            queue.addLast(message);
            emptyWaitSet.signal();
            return true;
        } finally {
            lock.unlock();
        }
    }

    /**
     * 将消息入队,设置超时时间,当队列为满时当前线程进入emptyWaitSet等待队列进行等待,若等待时间超过设置的超时时间则放弃入队<br/>
     * 将消息入队前先获取锁,获取失败则当前线程进入阻塞队列且被interrupt,直到其他线程释放锁当前线程再进行锁竞争<br/>
     * 消息入队成功时要随机唤醒一个在emptyWaitSet中等待的线程
     *
     * @param message  要入队的消息
     * @param timeout  超时时间
     * @param timeUnit 超时时间单位
     * @return 是否入队成功
     */
    public boolean push(Message message, Long timeout, TimeUnit timeUnit) {
        lock.lock();
        try {
            long nanos = timeUnit.toNanos(timeout);
            while (queue.size() == QUEUE_SIZE) {
                try {
                    if (nanos <= 0) {
                        return false;
                    }
                    nanos = fullWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            queue.addLast(message);
            emptyWaitSet.signal();
            return true;
        } finally {
            lock.unlock();
        }
    }
}
